package com.guass.nettyprotostuff;

import com.alibaba.fastjson.JSON;
import com.guass.nettyprotostuff.conf.ServerProperties;
import com.guass.nettyprotostuff.conf.ZKProperties;
import com.guass.nettyprotostuff.disruptor.CommandConsumer;
import com.guass.nettyprotostuff.disruptor.RingBufferWorkerPoolFactory;
import com.guass.nettyprotostuff.rpc.client.ClientTest;
import com.guass.nettyprotostuff.rpc.server.Server;
import com.guass.nettyprotostuff.rpc.server.ServerTest;
import com.guass.nettyprotostuff.service.MqService;
import com.guass.nettyprotostuff.service.RpcService;
import com.guass.nettyprotostuff.zookeeper.DataInfo;
import com.guass.nettyprotostuff.zookeeper.ZkConnect;
import com.guass.nettyprotostuff.zookeeper.ZookeeperService;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * @author guass
 */
@SpringBootApplication
@EnableScheduling
@Slf4j
public class NettyProtostuffApplication implements ApplicationListener<ApplicationEvent>, ZkConnect {

    @Resource
    private RpcService rpcService;

    @Resource
    private ServerProperties serverProperties;


    @Resource
    ZookeeperService zookeeperService;

    @Resource
    private Executor disruptorExecutor;

    @Resource
    private DefaultMQProducer mqProducer;

    @Resource
    private MqService mqService;

    @Resource
    private StringRedisTemplate stringRedisTemplate;


    public static void main(String[] args) {
        SpringApplication.run(NettyProtostuffApplication.class, args);
    }

    @Override
    public void onApplicationEvent(ApplicationEvent event) {
        if (event instanceof ApplicationReadyEvent){
            log.info("guass the boot ok!! {}",mqProducer);
            int mode = serverProperties.getMode();

            CompletableFuture.runAsync(()->{

                if (mode == 1) {
                    new ServerTest().startServerTest(rpcService,
                        serverProperties,
                        serverProperties.getTestStartPort(),serverProperties.getTestEndPort());
                }else if(mode == 2){
          //          new ClientTest(serverProperties.getIp()).start(serverProperties.getTestClientNum(), serverProperties.getTestStartPort(),serverProperties.getTestEndPort());
                }else {
                    new Server().startServer(rpcService,
                        serverProperties,
                        serverProperties.getRpcPort());
                }
            }, Executors.newSingleThreadExecutor());

            // TODO: 2022/7/27 启动zk 
//            zookeeperService.start(this);
            
            if (mode != 2) {
                initDisruptor();
                try {
                    String srvAddr = serverProperties.getMqNameSrvAddr();
                    if (!StringUtils.isEmpty(srvAddr)) {
                        mqProducer.start();
                        mqService.setMqStatus(true);
                    }
                } catch (MQClientException e) {
                    e.printStackTrace();
                }
            }

        }else if(event instanceof ContextClosedEvent){
            log.info("guass the application over!!!! ");
            mqProducer.shutdown();
        }
    }

    private void initDisruptor(){
        CommandConsumer[] commandConsumers = new CommandConsumer[serverProperties.getConsumerCount()];

        for (int i = 0; i < commandConsumers.length; i++) {
            commandConsumers[i] = new CommandConsumer("C" +i,rpcService,
                    serverProperties.getRingBufferSize(),
                    serverProperties.getRingSleepTime(),
                    serverProperties.getRingDoCount());
        }
        RingBufferWorkerPoolFactory.getInstance().initAndStart(ProducerType.MULTI,
                serverProperties.getRingBufferSize(),
                new YieldingWaitStrategy(),
                commandConsumers,
                disruptorExecutor);
    }

    @Override
    public void onConnected() {
        log.info("guass zk connected ok !!!");


        int mode = serverProperties.getMode();
        if (mode == 2) {
            new ClientTest(zookeeperService.getServerIp(),zookeeperService).start(
                    serverProperties.getTestClientNum(),
                    serverProperties.getTestStartPort(),
                    serverProperties.getTestEndPort());
        }
    }
}
